
/*
 * =====================================================================================
 *       Filename:  SessionCache.cc
 *
 *    Description:
 *
 *
 *        Version:  1.0
 *        Created:  03/06/19 15:59:09
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  liushuyong (), shuyong1999@163.com
 *        Company:  shuyong
 *
 * =====================================================================================
 */

#include "SessionCache.h"

DECLARE_int32(reqid_clean_interval);
DECLARE_int32(reqid_cache_timeout);
DECLARE_int32(max_sessions_count);

SessionCache* SessionCache::mgpInstance = new SessionCache();   //单例模式

SessionCache::SessionCache() {
#ifdef Debug
    LOG(INFO) << "New" << endl;
#endif
    mTimerID = 0;
    mSessionCount = 0;
    clean();
}

SessionCache::~SessionCache() {
#ifdef Debug
    LOG(INFO) << "Del" << endl;
#endif
}

SessionCache* SessionCache::getInstance() {
    return mgpInstance;
}

/**
 * @brief 将req 连接到某一session上，或者是创建一个新的， 但是没有初始化session中的数据
 *
 * @param reqid
 * @param seqid
 * @param errNo
 *
 * @return
 */
//RequestSessionSPtr SessionCache::fetchSession(const string& reqid, const int& seqid, int& errNo) {
RequestSessionSPtr SessionCache::attachSession(HttpParamSPtr param, int& errNo, bool& isNew) {
    int64_t timeNow = butil::gettimeofday_s();
    int32_t zoneID = getZoneIndex(param->mReqID);   //得到 zone
    //LOG(INFO) << "SessCount=" << mSessionCount << ":Attach" << endl;
    RWLockGuard zone_guard(mZoneLock[zoneID], B_WLOCK); //获取对用zone的读写锁 在退出时解锁
    SessionMap::iterator it = mSessZones[zoneID].find(param->mReqID);

    if (mSessZones[zoneID].end() != it) {   //reqID 对应的session 是否存在, 并更新时间
        if (param->mSeqID > 1) {
            isNew = false;

            if ((it->second.first)->mlastSeq + 1 != param->mSeqID) {    //TODO 检验Seq的排序问题
                LOG(ERROR) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ":seq err" << endl;
                errNo = ERROR_SEQID_ORDER;  //请求顺序有错误
                return NULL;
            } else {
                (it->second.first)->mlastSeq = param->mSeqID;
            }

#ifdef  Debug
            //LOG(INFO) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ",zoneID=" << zoneID << ",SessCount=" << mSessionCount << ":Attach" << endl;
#endif
            it->second.second = timeNow;    //更新时间
            return it->second.first;    //RequestSessionSPtr
        } else {   //是第一包，但是已经有session 了， 都不符合程序的逻辑
            LOG(ERROR) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ":seq err" << endl;
            errNo = ERROR_SEQID_FOR_SESSION;
            return NULL;
        }
    }

    //如果seeion没有, 但是seqID的值是第二包以后的数据，表明当前的请求超时
    if (param->mSeqID > 1) {
        LOG(ERROR) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ":timeout" << endl;
        errNo = ERROR_SESSION_TIMEOUT;
        return NULL;
    } else {
        if (mSessionCount >= FLAGS_max_sessions_count) {    //默认是100
            LOG(ERROR) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ",SessCount=" << mSessionCount << ":over" << endl;
            errNo = ERROR_SERVER_BUSY;
            return NULL;
        }

        //为第一包的请求创建一个session
#ifdef Debug
        //LOG(INFO) << "req=" << param->mReqID << ",seq=" << param->mSeqID << ",zoneID=" << zoneID << ",SessCount=" << mSessionCount << ":New" << endl;
#endif
        RequestSessionSPtr newSession(new RequestSession(param->mEngineParam, (param->mSeqID < 0) ? true : false));   //设置流式和非流式
        mSessZones[zoneID].insert(SessionMap::value_type(param->mReqID, make_pair(newSession, timeNow)));
        isNew = true;
        mSessionCount++;
        newSession->mlastSeq = param->mSeqID;   //
        //LOG (INFO) << "req=" << param->mReqID << ",useCount=" << newSession.use_count() << endl;  //useCount == 2
        return newSession;  //返回时，引用计数进行了递增操作。 newSession离开作用于 -1， 故返回后仍然为2
    }
}

/**
 * @brief TODO 删除指定session
 *
 * @param reqid
 *
 * @return
 */
//[ key:reqid   value->[key:RequestSessionSPtr  value:time] ]
int32_t SessionCache::delSession(const string& reqid) {
    int32_t zoneID = getZoneIndex(reqid);
    //LOG(INFO) << "delSession req0=" << reqid << ", mSessionCount=" << mSessionCount << endl;
    RWLockGuard zone_guard(mZoneLock[zoneID], B_RLOCK); //加锁
    //迭代器   ,没有找到， 返回end
    SessionMap::iterator it = mSessZones[zoneID].find(reqid);

    if (mSessZones[zoneID].end() != it) {
        RequestSessionSPtr sessSPtr = it->second.first;
        LOG(INFO) << "delSession req=" << reqid << ", mSessionCount=" << mSessionCount - 1 << endl;
        mSessZones[zoneID].erase(it);   //从map中移除 , 减少一次引用在RequestSession中释放
        mSessionCount--;
    } else {
        LOG(ERROR) << "reqid=" << reqid << ", session not found" << endl;
        return -1;
    }

    return 0;
}

int32_t SessionCache::getZoneIndex(const string& reqid) {
    int32_t ret = 0;

    if (reqid.size() <= 2) {
        return ret;
    }

    int32_t s1 = reqid[0];
    int32_t s2 = reqid[1];
    ret = (s1 + s2) % ZONE_NUM;
    return ret;
}

/**
 * @brief 定时器  查找超时session  TODO
 *
 * @param arg
 */
void SessionCache::cleanProc(void* arg) {
    //boost::ignore_unused(arg);
    SessionCache::getInstance()->clean();
}

/**
 * @brief
 *
 * @return
 */
int32_t SessionCache::clean() {
    deleteTimer();
    int clean_num = 0;

    //LOG(INFO) << "cleanSession 1req=" << ", mSessionCount=" << mSessionCount  <<  endl;

    for (int i = 0; i < ZONE_NUM; ++i) {
        RWLockGuard zone_guard(mZoneLock[i], B_RLOCK);
        int64_t timeNow = butil::gettimeofday_s();

        for (SessionMap::iterator it = mSessZones[i].begin(); it != mSessZones[i].end();) {
            int64_t elapsed = timeNow - it->second.second;  //消耗的时间

            //LOG(INFO) << "cleanSession 2req=  mSessionCount=" << mSessionCount  <<  endl;

            //if (elapsed > 10) {
            if (elapsed > FLAGS_reqid_cache_timeout) {
                LOG(INFO) << "cleanSession req=" << it->first << ", delay:" << timeNow - it->second.second << ", del" << ", mSessionCount=" << mSessionCount << endl;
                it->second.first->mspTTSEngineMgr.reset();  //释放engine
                mSessZones[i].erase(it++);  //移除map, 减少一次引用释放
                clean_num++;
                mSessionCount--;
            } else {
                it++;
            }
        }
    }

    //LOG(INFO) << "cleanSession 1req=" << ", mSessionCount=" << mSessionCount  <<  endl;
    //添加定时器加测Session超时
    registeTimer();
    return 0;
}

void SessionCache::registeTimer() {
    MutexLockGuard lockGuard(mTimerMutex);

    //LOG(ERROR) << "cleanSession registeTimer:" << ", mSessionCount=" << mSessionCount  <<  endl;

    if (0 != mTimerID) {
        LOG(ERROR) << "timerID:" << mTimerID << ":has registered" << endl;
        return;
    }

    timespec abstime = butil::seconds_to_timespec(butil::gettimeofday_s() + FLAGS_reqid_clean_interval);
    int32_t ret = bthread_timer_add(&mTimerID, abstime, cleanProc, NULL);

    if (0 != ret) {
        LOG(ERROR) << "cleanSession timer_id:" << mTimerID << ",ret:" << ret << ":bthread_timer_add failed." << endl;
    }
}

void SessionCache::deleteTimer() {
    MutexLockGuard lockGuard(mTimerMutex);

    //LOG(ERROR) << "cleanSession deleteTimer:" << ", mSessionCount=" << mSessionCount  <<  endl;

    if (0 != mTimerID) {
        bthread_timer_del(mTimerID);
        mTimerID = 0;
    }
}
